package com.thread;


import java.util.Objects;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 自定义线程池
 *
 * @author ljj
 */
public class MyThreadPoolExecutor {
	private final int corePoolCount;
	private final int maxPoolCount;
	private final int keepAliveTime;
	private final BlockingQueue<Runnable> workQueue;
	private final RejectedExecutionHandler handler;
	private static RejectedExecutionHandler defaultHandler = new DiscardPolicy();
	
	// 工作线程数量
	private final AtomicInteger workerCount = new AtomicInteger(0);
	// 保存所有 worker,用来管理线程的
	private final ConcurrentHashMap<Worker, Object> workers = new ConcurrentHashMap<>(16);
	
	
	// 几个变量数据
	static final AtomicInteger poolNumber = new AtomicInteger(0);
	static final AtomicInteger threadNumber = new AtomicInteger(0);
	static ThreadGroup group = new ThreadGroup("MyThreadPoolExecutor");
	
	
	// 线程池状态
	private final AtomicInteger poolStatus = new AtomicInteger(RUNNING);
	
	// 定义3个线程池状态，运行中，关闭和已终止
	private static final int RUNNING = 0;
	private static final int SHUTDOWN = 1;
	private static final int STOP = 2;
	
	
	private static boolean isRunning(int status) {
		return status == RUNNING;
	}
	
	public boolean isShutdown() {
		return !isRunning(poolStatus.get());
	}
	
	public BlockingQueue<Runnable> getWorkQueue() {
		return this.workQueue;
	}
	
	
	/**
	 * cas 方案处理工作线程数量的几个方法
	 */
	private boolean compareAndIncrementWorkerCount(int except) {
		return workerCount.compareAndSet(except, except + 1);
	}
	
	private boolean compareAndDecrementWorkerCount(int except) {
		return workerCount.compareAndSet(except, except - 1);
	}
	
	private void incrementWorkerCount() {
		do {
		} while (!compareAndIncrementWorkerCount(workerCount.get()));
	}
	
	private void decrementWorkerCount() {
		do {
		} while (!compareAndDecrementWorkerCount(workerCount.get()));
	}
	
	
	// 构造方法，初始化内部变量，用已缓存的默认策略
	public MyThreadPoolExecutor(int coreThreadCount, int maxPoolCount, int keepAliveTime, BlockingQueue<Runnable> workQueue) {
		this(coreThreadCount, maxPoolCount, keepAliveTime, workQueue, defaultHandler);
	}
	
	// 构造方法
	public MyThreadPoolExecutor(int corePoolCount, int maxPoolCount, int keepAliveTime, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
		if (corePoolCount < 0 || maxPoolCount <= 0 || corePoolCount > maxPoolCount) {
			throw new RuntimeException("线程池参数不正确");
		}
		if (workQueue == null || handler == null) {
			throw new NullPointerException();
		}
		this.corePoolCount = corePoolCount;
		this.maxPoolCount = maxPoolCount;
		this.keepAliveTime = keepAliveTime;
		this.workQueue = workQueue;
		this.handler = handler;
	}
	
	
	// 线程池的 execute 方法 ，还有个 submit 方法 ，一个用来提交不需要返回值的，一个提交需要返回值的
	public void execute(Runnable task) {
		if (task == null) {
			throw new NullPointerException("任务不能为null");
		}
		
		// 工作线程数 < 核心池数量,则尝试新增工作线程
		int workers = workerCount.get();
		if (workers < corePoolCount) {
			if (addWorker(task, true)) {
				// 成功直接返回
				return;
			}
			// 如果失败，存在一下可能，新增工作线程时核心线程刚好满了等等，或者线程池已经关闭
		}
		
		// 如果线程池还在运行中，且成功插入队列
		if (isRunning(poolStatus.get()) && workQueue.offer(task)) {
			// 再获取一次工作线程数 , 虽然已经放入队列，但还是需要再检查一次，因为可能有线程退出，核心线程数空出来了 或者线程池关闭了
			workers = workerCount.get();
			
			// 如果线程池已经没有运行了，则尝试移除掉这个任务
			if (!isRunning(poolStatus.get()) && remove(task)) {
				// 成功从队列中移除，说明这个任务没有被其他工作线程拿走，执行下拒绝策略
				reject(task);
			} else if (workers < corePoolCount) {
				// 线程数空出来了，则直接新增工作线程,这里任务给null任务，新线程直接从队列里拿任务，而不是非要执行当前的任务
				addWorker(null, true);
			}
		} else {
			// 走到这里，可能线程池关闭了，也可能队列满了无法插入
			// 那么我们最后再尝试一次新增工作线程跑任务，还失败的话就可以拒绝掉了
			if (!addWorker(task, false)) {
				reject(task);
			}
		}
	}
	
	/**
	 * 拒绝任务执行
	 */
	private void reject(Runnable task) {
		handler.rejectedExecution(task, this);
	}
	
	/**
	 * 从队列中尝试移除任务
	 */
	private boolean remove(Runnable task) {
		return workQueue.remove(task);
	}
	
	// 添加工作线程, isCore 说明是否是核心线程
	private boolean addWorker(Runnable firstTask, boolean isCore) {
		// 涉及到对线程池状态和线程数量的读取
		
		for (; ; ) {
			int status = poolStatus.get();
			int c = workerCount.get();
			
			// 检查线程池状态
			if (status != RUNNING) {
				return false;
			}
			
			// 检查任务是否为null,且队列也空，那么没必要新开工作线程
			if (firstTask == null && workQueue.isEmpty()) {
				return false;
			}
			
			// 按照是否核心线程，判断当前的线程数量是否超过了 核心数 或者 最大数，超过直接返回false
			if (c >= (isCore ? corePoolCount : maxPoolCount)) {
				return false;
			}
			
			// 现在开始自增下线程数量，若自增失败，则表示存在其他的并发线程访问，那么重新循环回来，下一轮循环再次检查状态和线程数量
			if (compareAndIncrementWorkerCount(c)) {
				break;
			}
		}
		
		// 到这里，成功增加了线程数量，准备开始正式开工作线程
		// 线程不一定能新建成功，比如系统资源耗尽,所以try  finally 中写一下添加工作线程失败逻辑,同时再次检查线程池状态
		Worker worker = null;
		boolean addWorker = false;
		try {
			worker = new Worker(firstTask);
			final Thread thread = worker.thread;
			if (thread != null) {
				int status = poolStatus.get();
				
				if (status == RUNNING) {
					workers.put(worker, 1);
					
					// 启动线程
					thread.start();
					addWorker = true;
				}
			}
		} finally {
			// 如果新增工作线程失败，则调用失败逻辑
			if (!addWorker) {
				addWorkerFailed(worker);
			}
		}
		return addWorker;
	}
	
	/**
	 * 新增工作线程失败逻辑,删除掉保存的工作线程引用并减少当前线程数
	 */
	private void addWorkerFailed(Worker worker) {
		if (worker != null) {
			workers.remove(worker);
		}
		decrementWorkerCount();
	}
	
	/**
	 * 实际运行的工作线程任务，而不是直接跑 Runnable 里的run
	 */
	private void runWorker(Worker worker) {
		Thread thread = Thread.currentThread();
		Runnable task = worker.firstTask;
		worker.firstTask = null;
		
		// 设置工作线程是否出现异常问题，默认true
		boolean workerCompleteExcept = true;
		
		try {
			while (task != null || (task = getTask()) != null) {
				// 每次执行任务前都检查下当前线程池的状态,只要不是stop就行，一旦是stop，就立即打断线程
				if (poolStatus.get() == STOP && !thread.isInterrupted()) {
					thread.interrupt();
				}
				
				try {
					task.run();
				} finally {
					// 记得清空掉task ，否则下一轮循环跑的还是之前的任务
					task = null;
					worker.completeTasks++;
				}
			}
			// 循环任务正常退出，就把这个标志设置为false,表示任务没有异常
			workerCompleteExcept = false;
		} finally {
			// 处理下线程退出的事情
			processWorkerExit(worker, workerCompleteExcept);
		}
	}
	
	/**
	 * 处理工作线程退出的逻辑
	 * 如果是正常退出
	 */
	private void processWorkerExit(Worker worker, boolean workerCompleteExcept) {
		// 减一下工作线程数
		decrementWorkerCount();
		
		// 工作线程集合中先移除
		workers.remove(worker);
		
		// 检查下线程池状态 和 线程数量，看看有没有需要重新建立线程
		if (poolStatus.get() != STOP) {
			/*
			 * 这里描述下大概逻辑，即每个线程试图退出都尽可能检查下并且重新开工作线程
			 *
			 * 如果正常退出，则检查核心线程数量  不够就重新尝试起线程
			 * 异常退出则总是尝试起线程
			 */
			if (!workerCompleteExcept) {
				int c = workerCount.get();
				if (c >= corePoolCount) {
					// 线程数超过核心线程数，那就没必要开线程了，直接return 结束掉
					return;
				}
			}
			addWorker(null, false);
		}
	}
	
	// 获取任务，分核心和非核心，核心一直等待，非核心等一段时间
	private Runnable getTask() {
		boolean timedOut = false;
		
		while (true) {
			int c = workerCount.get();
			
			int status = poolStatus.get();
			// 如果是 shutdown 状态，且队列已经空，则直接返回null
			// 或者线程池已经 stop ，则直接返回null
			if (status == SHUTDOWN && workQueue.isEmpty()) {
				return null;
			} else if (status == STOP) {
				return null;
			}
			
			// 一旦队列为空，且 线程数 > 核心线程数  或者 当前线程已经获取任务超时一次了，则直接返回null
			if (workQueue.isEmpty() && (c > corePoolCount || timedOut)) {
				return null;
			}
			
			try {
				Runnable task = c > corePoolCount ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take();
				if (task != null) {
					return task;
				}
				timedOut = true;
			} catch (InterruptedException e) {
				timedOut = false;
			}
		}
	}
	
	// 关闭线程池，打断空闲线程
	public synchronized void shutdown() {
		poolStatus.set(SHUTDOWN);
		
		for (Worker worker : workers.keySet()) {
			Thread.State state = worker.thread.getState();
			if (state == Thread.State.WAITING || state == Thread.State.TIMED_WAITING) {
				if (worker.thread.isAlive() && !worker.thread.isInterrupted()) {
					worker.thread.interrupt();
				}
			}
		}
	}
	
	// 关闭线程池，打断所有线程
	public synchronized void shutdownNow() {
		poolStatus.set(STOP);
		
		for (Worker worker : workers.keySet()) {
			if (worker.thread.isAlive() && !worker.thread.isInterrupted()) {
				worker.thread.interrupt();
			}
		}
	}
	
	
	private Thread newThread(Runnable runnable) {
		SecurityManager s = System.getSecurityManager();
		group = (s != null) ? s.getThreadGroup() :
				Thread.currentThread().getThreadGroup();
		Thread t = new Thread(group, runnable, "pool-0" + "-thread-" + threadNumber.getAndIncrement());
		t.setDaemon(false);
		t.setPriority(Thread.NORM_PRIORITY);
		return t;
	}
	
	
	private final class Worker implements Runnable {
		/**
		 * 持有实际线程
		 */
		private final Thread thread;
		/**
		 * 首个任务
		 */
		private Runnable firstTask;
		
		/**
		 * 完成任务计数
		 */
		volatile long completeTasks;
		
		public Worker(Runnable firstTask) {
			this.firstTask = firstTask;
			this.thread = newThread(this);
		}
		
		@Override
		public void run() {
			// runWorker是线程池的方法，因为runWorker中要用到线程池的一些变量等，不能作为自己的私有方法,否则访问不到线程池
			runWorker(this);
		}
		
		@Override
		public boolean equals(Object o) {
			if (this == o) {
				return true;
			}
			if (o == null || getClass() != o.getClass()) {
				return false;
			}
			Worker worker = (Worker) o;
			return Objects.equals(thread, worker.thread);
		}
		
		@Override
		public int hashCode() {
			return Objects.hash(thread);
		}
	}
	
	public interface RejectedExecutionHandler {
		/**
		 * 拒绝执行方案
		 */
		void rejectedExecution(Runnable r, MyThreadPoolExecutor executor);
	}
	
	/**
	 * 抛出异常方案
	 */
	public static class AbortPolice implements RejectedExecutionHandler {
		@Override
		public void rejectedExecution(Runnable r, MyThreadPoolExecutor executor) {
			throw new RuntimeException("任务 " + r.toString() + " 被线程池 " + executor.toString() + " 拒绝执行");
		}
	}
	
	/**
	 * 忽略方案,啥也不干
	 */
	public static class DiscardPolicy implements RejectedExecutionHandler {
		@Override
		public void rejectedExecution(Runnable r, MyThreadPoolExecutor executor) {
			// ignore
			System.out.println("任务 " + r.toString() + " 被线程池忽略!");
		}
	}
	
	/**
	 * 调用者你自己的线程执行好了
	 */
	public static class CallerRunsPolicy implements RejectedExecutionHandler {
		@Override
		public void rejectedExecution(Runnable r, MyThreadPoolExecutor executor) {
			if (executor.isShutdown()) {
				// 线程池都关闭了，就不管了，忽略任务
			} else {
				// 没关的话，就用调用 execute() 方法的那个调用者线程来直接跑 run() 方法
				r.run();
			}
		}
	}
	
	/**
	 * 丢弃一个队列中最老的任务（也就是队列的第一个任务），然后重新尝试执行下
	 */
	public static class DiscardOldestPolicy implements RejectedExecutionHandler {
		@Override
		public void rejectedExecution(Runnable r, MyThreadPoolExecutor executor) {
			if (!executor.isShutdown()) {
				// 从队列拿掉一个任务,然后重新尝试 execute()
				executor.getWorkQueue().poll();
				executor.execute(r);
			}
		}
	}
}
